package delayqueue

import (
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/ouqiang/delay-queue/config"
)

var (
	// 每个定时器对应一个bucket
	timers []*time.Ticker
	// bucket名称chan
	bucketNameChan <-chan string
)

// Init 初始化延时队列
func Init() {
	RedisPool = initRedisPool()
	initTimers()
	bucketNameChan = generateBucketName()
}

// Push 添加一个Job到队列中
func Push(job Job) error {
	if job.Id == "" || job.Topic == "" || job.Delay < 0 || job.TTR <= 0 {
		return errors.New("invalid job")
	}
	//将任务内容编码后插入到redis的字符串库中
	err := putJob(job.Id, job)
	if err != nil {
		log.Printf("添加job到job pool失败#job-%+v#%s", job, err.Error())
		return err
	}
	//将任务id和时间戳插入到有序集合中,方便以后快速查找
	err = pushToBucket(<-bucketNameChan, job.Delay, job.Id)
	if err != nil {
		log.Printf("添加job到bucket失败#job-%+v#%s", job, err.Error())
		return err
	}

	return nil
}

// Pop 轮询获取Job.是消费者的入口,从只读队列的左侧读取元素,直到没有数据为止开始阻塞
func Pop(topics []string) (*Job, error) {
	jobId, err := blockPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
	if err != nil {
		return nil, err
	}

	// 队列为空
	if jobId == "" {
		return nil, nil
	}

	// 获取job元信息
	job, err := getJob(jobId)
	if err != nil {
		return job, err
	}

	// 消息不存在, 可能已被删除
	if job == nil {
		return nil, nil
	}
	//计算超时的时间戳,并重新放入任务队列,等待二次处理
	timestamp := time.Now().Unix() + job.TTR
	//用bucket名做Key,加上超时和任务id,存入有序集合中.
	//以便服务器轮询时二次确认,若已经删除了任务json,会自动删除这个二次任务
	err = pushToBucket(<-bucketNameChan, timestamp, job.Id)

	return job, err
}

// Remove 删除Job
func Remove(jobId string) error {
	return removeJob(jobId)
}

// Get 查询Job
func Get(jobId string) (*Job, error) {
	job, err := getJob(jobId)
	if err != nil {
		return job, err
	}

	// 消息不存在, 可能已被删除
	if job == nil {
		return nil, nil
	}
	return job, err
}

// 轮询获取bucket名称, 使job分布到不同bucket中, 提高扫描速度
func generateBucketName() <-chan string {
	//创建一个无缓存的通道,仅在有协程接收bucket名时就生成一个
	c := make(chan string)
	go func() {
		i := 1
		for {
			c <- fmt.Sprintf(config.Setting.BucketName, i)
			if i >= config.Setting.BucketSize {
				i = 1
			} else {
				i++
			}
			fmt.Printf("轮询的bucket名称已生成:%v\n", time.Now())
		}
	}()

	return c
}

// 初始化定时器
func initTimers() {
	timers = make([]*time.Ticker, config.Setting.BucketSize)
	var bucketName string
	for i := 0; i < config.Setting.BucketSize; i++ {
		timers[i] = time.NewTicker(1 * time.Second)
		//bucket_name = dq_bucket_%d    ; bucket在redis中的键名, %d必须保留
		//拼接出带编号的数据桶名称
		bucketName = fmt.Sprintf(config.Setting.BucketName, i+1)
		go waitTicker(timers[i], bucketName)
	}
}

func waitTicker(timer *time.Ticker, bucketName string) {
	for {
		select {
		case t := <-timer.C:
			tickHandler(t, bucketName)
		}
	}
}

// 扫描bucket, 取出延迟时间小于当前时间的任务
func tickHandler(t time.Time, bucketName string) {
	for {
		bucketItem, err := getFromBucket(bucketName)
		if err != nil {
			log.Printf("扫描bucket错误#bucket-%s#%s", bucketName, err.Error())
			return
		}

		// 当集合为空后,再退出循环
		if bucketItem == nil {
			return
		}

		// 若碰到延迟时间未到的任务,就退出循环
		if bucketItem.timestamp > t.Unix() {
			return
		}
		//经过上面的排除后,剩下的就是已经过期的任务
		job, err := getJob(bucketItem.jobId)
		if err != nil {
			log.Printf("获取Job元信息失败#bucket-%s#%s", bucketName, err.Error())
			continue
		}

		//若任务json不存在, 就说明已经被消费者删除,这相当于二次确认的信号
		//所以就要从bucket中删除该任务id
		if job == nil {
			removeFromBucket(bucketName, bucketItem.jobId)
			continue
		}

		// 若任务json还存在,且delay晚于当前时间,就删除掉旧任务,并重新计算延迟时间后再次插入有序集合
		if job.Delay > t.Unix() {
			// 从bucket中删除旧的jobId
			removeFromBucket(bucketName, bucketItem.jobId)
			// 重新计算delay时间并放入bucket中
			pushToBucket(<-bucketNameChan, job.Delay, bucketItem.jobId)
			continue
		}
		//最后将任务ID插入到只读队列,等待消费者来使用
		err = pushToReadyQueue(job.Topic, bucketItem.jobId)
		if err != nil {
			log.Printf("JobId放入ready queue失败#bucket-%s#job-%+v#%s",
				bucketName, job, err.Error())
			continue
		}

		//如果都完成了,最后就从bucket中删除该任务
		removeFromBucket(bucketName, bucketItem.jobId)
	}
}
